package netty.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import netty.common.network.Message;
import netty.common.serializer.TcpDecoder;
import netty.common.serializer.TcpEncoder;
import test.PerformanceAnalyze;

public class TCPClient {

    static PerformanceAnalyze performanceAnalyze = new PerformanceAnalyze();

    private final String host = System.getProperty("serverHost", "127.0.0.1");

    private final int port = 9955;

    private ChannelFuture channelFuture;

    public void start() throws Exception {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(1);

        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_SNDBUF, 6000*1024)
                    .option(ChannelOption.SO_RCVBUF, 6000*1024)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 15000)
                    .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(512 * 1024, 1024 * 1024))
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
//                    ch.pipeline().addLast(new JdkZlibEncoder());
//                    ch.pipeline().addLast(new JdkZlibDecoder());
                    ch.pipeline().addLast(new TcpEncoder());
                    ch.pipeline().addLast(new TcpDecoder());
                    ch.pipeline().addLast(new ClientHandler());
                }
            });

            channelFuture = b.connect(host, port).sync();

//            channelFuture.channel().closeFuture().sync();
        } finally {
//            workerGroup.shutdownGracefully();
        }
    }

    public void sendMsg(Message msg, boolean needFlush) {
        long now = System.nanoTime();
        if (needFlush) {
            channelFuture.channel().writeAndFlush(msg).addListener(future -> {
                performanceAnalyze.delay(System.nanoTime() - now);
            });
        } else {
            channelFuture.channel().write(msg).addListener(future -> {
                performanceAnalyze.delay(System.nanoTime() - now);
            });
        }

        performanceAnalyze.meter();
        performanceAnalyze.length(msg.bodyLength());
    }

}
